package com.cat.paxos.proposer;

import com.cat.paxos.acceptor.ConfirmProposalResponse;
import com.cat.paxos.acceptor.ReceiveProposalResponse;
import com.cat.paxos.common.Response;
import com.cat.paxos.proposal.ProposalNumberComparator;

import java.math.BigInteger;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @auther Cat.wang
 * @date 2020/1/16 14:44
 */
public class HalfUpAcceptorStrategy implements AcceptorStrategy {

    public static final String DEFAULT_START_PROPOSAL_NUMBER = "0";

    private Comparator<String> proposalNumberComparator = ProposalNumberComparator.DEFAULT_PROPOSAL_NUMBER_COMPARATOR;

    @Override
    public CommitWaitConfirmProposalResponse decisionCommitWaitConfirmProposal(int total, List<ReceiveProposalResponse> responses) throws CommitProposalException {

        if (responses.size() < total / 2) {
            throw new CommitProposalException("Insufficient number of people to receive proposals");
        }

        boolean oked = responses.stream().filter(Response::ok).count() > total / 2;
        String proposalNumbered;
        if (oked) {
            proposalNumbered = responses.stream().filter(Response::ok).map(ReceiveProposalResponse::proposalNumber).max(proposalNumberComparator).orElse(null);
        } else {
            proposalNumbered = responses.stream().map(ReceiveProposalResponse::proposalNumber).max(proposalNumberComparator).orElse(null);
        }

        return new CommitWaitConfirmProposalResponse() {

            private boolean ok = oked;

            private String proposalNumber = proposalNumbered;

            @Override
            public String proposalNumber() {
                return this.proposalNumber;
            }

            @Override
            public boolean ok() {
                return this.ok;
            }
        };
    }

    @Override
    public CommitProposalResponse decisionCommitConfirmProposal(int total, List<ConfirmProposalResponse> responses) throws CommitProposalException {

        List<ConfirmProposalResponse> success = responses.stream().filter(Response::ok).collect(Collectors.toList());

        boolean oked = !success.isEmpty();
        String proposalNumbered;
        if (oked) {
            proposalNumbered = success.stream().map(ConfirmProposalResponse::proposalNumber).max(proposalNumberComparator).orElse(null);
        } else {
            proposalNumbered = responses.stream().map(ConfirmProposalResponse::proposalNumber).max(proposalNumberComparator).orElse(null);
        }

        return new CommitProposalResponse() {

            private boolean ok = oked;

            private String proposalNumber = proposalNumbered;

            @Override
            public String proposalNumber() {
                return proposalNumber;
            }

            @Override
            public boolean ok() {
                return this.ok;
            }
        };
    }

    @Override
    public String decisionProposalNumber(int total, List<String> proposalNumbers) throws PrepareProposalException {

        if (proposalNumbers.size() < total / 2) {
            throw new PrepareProposalException("Insufficient number of people to receive proposals");
        }

        String proposalNumber = proposalNumbers.stream().filter(Objects::nonNull).max(proposalNumberComparator).orElse(null);

        if (proposalNumber == null) {
            proposalNumber = DEFAULT_START_PROPOSAL_NUMBER;
        } else {
            proposalNumber = new BigInteger(proposalNumber).add(BigInteger.ONE).toString();
        }

        return proposalNumber;
    }
}
